package com.imooc.kafka.consumer;


import com.google.gson.Gson;
import com.imooc.kafka.common.MessageEntity;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.stereotype.Component;

@Slf4j
@Component
public class SimpleConsumer {

    private final Gson gson = new Gson();


    //消费两个topic
    @KafkaListener(topics ={"${kafka.topic.default}","mytopic"}, containerFactory = "kafkaListenerContainerFactory")

    public void receive(@Payload String payload,
                        @Header(KafkaHeaders.RECEIVED_TOPIC) String topic,
                        @Header(KafkaHeaders.RECEIVED_MESSAGE_KEY) Integer key,
                        @Header(KafkaHeaders.RECEIVED_PARTITION_ID) int partition,
                        @Header(KafkaHeaders.CONSUMER) KafkaConsumer consumer,
                        @Header(KafkaHeaders.OFFSET) String offset) {

        System.out.println("key:"+key);
        System.out.println("consumer:"+consumer.listTopics());
        System.out.println("partition:"+partition);
        System.out.println("payload:"+payload);
        System.out.println("offset:"+offset);
//        System.out.println(Thread.currentThread().getName()+message);
        //不能再配置中既配置kafka.consumer.enable-auto-commit=true 自动提交; 然后又在监听器中使用手动提交
//        ack.acknowledge();//手动提交偏移量
//        System.out.println("接收成功"+gson.toJson(message));
    }
}